package com.friendsystem.hadoop;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class FriendRecommendationJob {
    public static class FriendMapper extends Mapper<Object, Text, Text, Text> {
        public void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {
            // 输入格式: user1,user2 (表示好友关系)
            String[] friends = value.toString().split(",");
            if (friends.length == 2) {
                context.write(new Text(friends[0]), new Text(friends[1]));
                context.write(new Text(friends[1]), new Text(friends[0]));
            }
        }
    }

    public static class FriendReducer extends Reducer<Text, Text, Text, Text> {
        public void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            // 为每个用户生成好友推荐
            List<String> friends = new ArrayList<>();
            for (Text val : values) {
                friends.add(val.toString());
            }

            // 生成推荐: 好友的好友(排除已经是好友的)
            for (int i = 0; i < friends.size(); i++) {
                for (int j = i + 1; j < friends.size(); j++) {
                    context.write(new Text(friends.get(i)), new Text(friends.get(j)));
                    context.write(new Text(friends.get(j)), new Text(friends.get(i)));
                }
            }
        }
    }

    public Job createRecommendationJob(String userId) throws IOException {
        Configuration conf = new Configuration();
        conf.set("mapreduce.job.queuename", "default");
        conf.set("recommend.for.user", userId);

        Job job = Job.getInstance(conf, "friend recommendation");
        job.setJarByClass(FriendRecommendationJob.class);
        job.setMapperClass(FriendMapper.class);
        job.setReducerClass(FriendReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        FileInputFormat.addInputPath(job, new Path("/input/friendships"));
        FileOutputFormat.setOutputPath(job, new Path("/output/recommendations_" + System.currentTimeMillis()));

        return job;
    }

    public List<User> getRecommendationResults() {
        // 这里应该实现从HDFS读取推荐结果并解析为User对象列表
        // 简化示例，实际应从HDFS读取文件
        return new ArrayList<>();
    }
}